package com.atguigu.aggreagte;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author gmd
 * @desc 简单聚合（sum/min/max/minBy/maxBy）算子
 * @since 2024-11-25 17:23:26
 */
public class SimpleAggregate {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<WaterSensor> sensorDS = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s1", 11L, 11),
                new WaterSensor("s1", 8L, 8),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3)
        );

        KeyedStream<WaterSensor, String> sensorKS = sensorDS
                .keyBy(new KeySelector<WaterSensor, String>() {
                    @Override
                    public String getKey(WaterSensor value) throws Exception {
                        return value.getId();
                    }
                });

        /*
         * 简单聚合算子
         * ======================
         * 简单聚合算子用于对同一个key的数据进行聚合。
         *
         * 注意：
         * 1. 简单聚合算子只能在keyBy之后调用。
         * 2. 简单聚合算子适用于Tuple类型的数据，对POJO类型的数据不适用。
         */
        // 传位置索引的，适用于Tuple类型，POJO不行
//        SingleOutputStreamOperator<WaterSensor> result = sensorKS.sum(2);
        // 传字段名称，适用于POJO
//        SingleOutputStreamOperator<WaterSensor> result = sensorKS.sum("vc");


        /*
         * 简单聚合算子 - max 和 maxBy 的区别
         * -----------------------------------
         * max 和 maxBy 都是用于求最大值的算子，但它们的行为略有不同。
         *
         * max：只会取比较字段的最大值，非比较字段保留第一次的值。
         * maxBy：取比较字段的最大值，同时非比较字段取最大值这条数据的值。
         *
         * 例如，如果我们有以下数据：
         *   +----+-----+-----+
         *   | id | vc  | ts  |
         *   +----+-----+-----+
         *   | s1 | 1   | 1   |
         *   | s1 | 11  | 11  |
         *   | s2 | 2   | 2   |
         *   | s3 | 3   | 3   |
         *   +----+-----+-----+
         *
         * 使用 max("vc") 算子后，结果将是：
         *   +----+-----+-----+
         *   | id | vc  | ts  |
         *   +----+-----+-----+
         *   | s1 | 11  | 1   |
         *   | s2 | 2   | 2   |
         *   | s3 | 3   | 3   |
         *   +----+-----+-----+
         *
         * 使用 maxBy("vc") 算子后，结果将是：
         *   +----+-----+-----+
         *   | id | vc  | ts  |
         *   +----+-----+-----+
         *   | s1 | 11  | 11  |
         *   | s2 | 2   | 2   |
         *   | s3 | 3   | 3   |
         *   +----+-----+-----+
         */
//        SingleOutputStreamOperator<WaterSensor> result = sensorKS.max("vc");
//        SingleOutputStreamOperator<WaterSensor> result = sensorKS.min("vc");
        SingleOutputStreamOperator<WaterSensor> result = sensorKS.maxBy("vc");
//        SingleOutputStreamOperator<WaterSensor> result = sensorKS.minBy("vc");

        result.print();
        env.execute();
    }

    /*
     * maxBy 的工作机制
     * maxBy("vc") 是一个Flink中的聚合操作，用于查找指定字段（在这里是 "vc"）上的最大值。具体行为是：
     *
     * 按键分组：keyBy 按 WaterSensor 的 id 进行分组。因此，相同 id 的元素会被逻辑上分配到同一组进行处理。
     * 逐条处理：Flink的数据流是实时处理的，maxBy 操作是逐条计算的。当每条新数据到达时，它会与当前分组的最大值进行比较和更新。
     * 保持最新的最大值：maxBy 会维护一个“状态”，这个状态包含每个 Key 的当前最大值。如果新数据的 vc 值比当前的最大值大，则更新该状态。
     */

}
